package com.atguigu.edu.realtime.app.dwd.db;

import com.atguigu.edu.realtime.utils.HbaseUtil;
import com.atguigu.edu.realtime.utils.MyKafkaUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;
//考试
public class DwdTestExamination {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(5);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        tableEnv.executeSql(MyKafkaUtil.getTopicDbDDL("DwdTestExamination"));
        tableEnv.getConfig().setIdleStateRetention(Duration.ofSeconds(10));

        Table test_exam = tableEnv.sqlQuery("select `data`['user_id'] user_id,`data`['score'] score,`data`['duration_sec'] duration_sec,`data`['paper_id'] paper_id,proc_time from topic_db where `table`='test_exam'");
        tableEnv.createTemporaryView("test_exam", test_exam);

        Table test_exam_question = tableEnv.sqlQuery("select `data`['question_id'] question_id,`data`['exam_id'] exam_id,`data`['score'] score,`data`['paper_id'] paper_id,`data`['is_correct'] is_correct from topic_db where `table`='test_exam_question'");
        tableEnv.createTemporaryView("test_exam_question", test_exam_question);

        tableEnv.executeSql(HbaseUtil.getBaseDicDDL());


        Table examination = tableEnv.sqlQuery("select te.user_id user_id,te.paper_id paper_id,te.score score,te.duration_sec duration_sec,teq.question_id question_id,teq.exam_id exam_id,teq.score question_score,teq.is_correct is_correct,tp.course_id course_id from test_exam te join test_exam_question teq\n" +
                "on te.paper_id = teq.paper_id join test_paper for system_time as of te.proc_time as tp on te.paper_id = tp.id");

        tableEnv.createTemporaryView("examination", examination);


        tableEnv.executeSql("create table dwd_test_examination (\n" +
                "user_id string,paper_id string,score string,duration_sec string,question_id string,exam_id string,question_score string,is_correct string,course_id string,PRIMARY KEY (user_id) NOT ENFORCED)"
                + MyKafkaUtil.getUpsertKafkaDDL("dwd_test_examination"));


        tableEnv.executeSql("insert into dwd_test_examination select * from examination");


    }
}
